/*
 * LowLevelDispatcher.java    0.0.1    19/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs.server.impl;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.mentalsmash.tazio.net.cs.server.ProtocolHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Dispatcher's implementation for nonblocking socket-level networking.
 * <p>
 * On readiness events on connected sockets, it runs the appropriate handler using an 
 * {@link java.util.concurrent.Executor Executor} passed at construction time.
 * </p>
 * <p>
 * This class and all the classes in this package are largely inspired to the presentation
 * "How To Build A Scalable NIO Server, Mark II" by Ron Hitchens (http://ronsoft.com)
 * </p>
 * @author     Ron Hitchens - original work  
 * @author     Andrea Reale <andrea.reale@gmail.com> - adaptation
 * @version    0.0.1    19/set/2009
 * @see        org.mentalsmash.tazio.net.cs.server.impl.Dispatcher Dispatcher
 */
public class LowLevelDispatcher implements Dispatcher, Runnable {

    private final static Logger log = LoggerFactory.getLogger(LowLevelDispatcher.class);
    private final static int WAITING_QUEUE_SIZE = 100;
    private final static String THREAD_NAME = "Dispatcher";

    /*
     * the listeners registered on the acceptors. Accesses to this set are synchronised using the set's lock.
     */
    private final Set<DispatcherListener> registeredListeners = new HashSet<DispatcherListener>();

    /**
     * This lock is used to make selection process wait before starting, until possibly registering
     * (or unregistering) handler threads complete their registration (deregistration)
     */
    private final ReadWriteLock guard = new ReentrantReadWriteLock(); 
    private final Executor executor; // The one who manages to start and run handler threads
    private Selector selector;
    private final BlockingQueue<HandlerAdapter> waitingForStatusChange; // Handlers who wish to change their interests
    // are inserted in this waiting queue, and 
    // their status gets changes after the current
    // selection process has returned.

    private final BufferPool bufferPool;
    private volatile boolean dispatching = false;


    private AtomicInteger numRequestsServed = new AtomicInteger(); //monitors the number of requests served by the dispatcher.


    /**
     * Create a new <code>LowLevelDispatcher</code> with the specified values.
     * @param e an instance of executor. It will be used to run handlers' threads
     * @param pool A pool of direct buffers.
     * @throws IOException if an IO exception occurs while retrieving a selector for the dispatcher
     */
    public LowLevelDispatcher(Executor e, BufferPool pool) throws IOException {
	if((e == null) || (pool == null)){
	    log.error("Executor or BufferPool are null");
	    throw new IllegalArgumentException("Null pointers in constructor");
	}

	this.executor = e;
	this.bufferPool = pool;

	waitingForStatusChange = new ArrayBlockingQueue<HandlerAdapter>(WAITING_QUEUE_SIZE);

	try {
	    selector = Selector.open();
	} catch(IOException e1) {
	    log.error("Unable to open the selector",e1);
	    throw e1;
	}

    }

    /**
     * Creates a thread and makes it run the dispatcher.
     * @return the thread that is running the dispatcher.
     */
    //CONCURRENCY: Called by external thread.
    public synchronized Thread start(){

	Thread dispatcherThread = new Thread(this);
	dispatcherThread.setName(THREAD_NAME);
	dispatcherThread.start();

	while(!dispatching) {
	    try {
		wait();
	    } catch (InterruptedException e) {
		log.error("Unexpected Interruption on the thread while starting the dispatcher " , e);
		continue;
	    }
	}

	//notify the listeners
	notifyStart();


	log.debug("Dispatcher Thread started");
	return dispatcherThread;
    }


    /**
     * Notifies the listeners that the dispatcher has started.
     */
    private void notifyStart() {
	synchronized (registeredListeners) {
	    for(DispatcherListener l : registeredListeners) {
		l.dispatcherStarted();
	    }
	}
    }
    /*
     * -- Runnalble interface implementation --
     */

    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
	try {
	    synchronized (this) {
		if(!selector.isOpen()) {
		    log.debug("Selector was closed. Opening a new one.");
		    selector = Selector.open();
		}
	    }


	    dispatching = true;

	    synchronized (this) {
		notifyAll(); //notify the thread that started the dispatcher, that the dispatcher has started.
	    }

	    dispatch();
	} catch (IOException e) {
	    log.error("Unexpected IOException while dispatching",e);
	    this.stop();
	    notifyStop(e);
	} catch(Throwable e) { //Any other unexpected exception that otherwise no one will catch
	    log.error("Unexpected Exception while dispatching", e);
	    this.stop();
	    notifyStop(e);
	}

	clearDispatcher();

    }

    /**
     * Notifies every listener that the dispatcher has stopped.
     * This method is called by the dispatcher thread
     * @param e the cause of the stop
     */
    private void notifyStop(final Throwable e) {

	synchronized (registeredListeners) {
	    for(DispatcherListener l : registeredListeners) {
		l.dispatcherStopped(e);
	    }
	}

    }

    /**
     * Clears the registrations to the selctor and closes it.
     */
    /*
     * Package-level visibility because this method could also be used for testing purposes
     */
    void clearDispatcher() {

	// The second should be better
	// Set<SelectionKey> ready = selector.selectedKeys();
	Set<SelectionKey> ready = selector.keys();

	//unregistering keys that where selected but not completely served.
	for(SelectionKey key : ready) {
	    HandlerAdapter adapter = (HandlerAdapter)key.attachment();
	    unregister(adapter);
	}

	try {
	    selector.close();
	    log.debug("Selector closed. " + THREAD_NAME + " ended.");
	} catch (IOException e) {
	    log.error("Unexpected IOException while closing the selector",e);
	}
    }

    /*
     * -- Dispatcher interface implementation --
     */

    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Dispatcher#dispatch()
     */
    //CONCURRENCY: called by the Dispatcher Thread only

    public void dispatch() throws IOException {
	// dispatching is volatile, therefore it is guaranteed that reads and writes to it
	// by different threads are in happened-before order
	while(dispatching) { 
	    Set<SelectionKey> selected = null; 

	    selectorBarrier(); // see selectorBarrier doc
	    selector.select();

	    checkStatusChangeQueue(); // check whether any adapter is dead or it has completed its work.
	    // In the second case, set its interestOps (that were previoysly)
	    // reset.

	    selected = selector.selectedKeys();

	    for (SelectionKey key :selected ){
		HandlerFutureTask toRun = null; //the task to run for the current key

		HandlerAdapter adapter = (HandlerAdapter)key.attachment();

		adapter.prepareToRun();
		adapter.key().interestOps(0); //Do not notify anything while this adapter is running
		toRun = new HandlerFutureTask(adapter);
		executor.execute(toRun);
	    }

	    selected.clear(); //empty the set
	}
    }



    /**
     * Selector barrier: the next two calls have the purpose to make the Dispatcher thread
     * wait until threads registering their channels release their read lock. In other words
     * - before selection - the dispatcher waits for pending registrations to complete. 
     */
    //CONCURRENCY: This is called only by the dispatcher thread
    private void selectorBarrier() {
	log.debug("Wating on the barrier.");
	guard.writeLock().lock();
	guard.writeLock().unlock();
	log.debug("Out of the barrier.");
    }

    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Dispatcher#stop()
     */
    @Override
    //CONCURRENCY: this could be get called by anyone, but dispatching is volatile
    public synchronized void stop() {
	log.debug("Stopping the dispatcher");
	dispatching = false;
	selector.wakeup();


    }

    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Dispatcher#register(java.nio.channels.SelectableChannel, org.mentalsmash.tazio.net.cs.server.ProtocolHandler)
     */
    //CONCURRENCY: This method is called by any thread who wants to register a channel, but does not need 
    //             special synchornization since the selector is internally thread-safe. The sharedLock is
    //             taken to make the dispatcher wait for registration completion, before resuming the 
    //             selection
    @Override
    public ChannelFacade register(SelectableChannel chan, ProtocolHandler handler)
    throws IOException {

	if(!isDispatching()){
	    String msg = "Unable to register: the Dispatcher is not running";
	    log.warn(msg);
	    throw new DispatcherNotRunningException(msg);
	}

	HandlerAdapter adapter = new HandlerAdapter(handler, this, this.bufferPool);
	chan.configureBlocking(false);

	adapter.registering(); //notifies the adapter

	acquireSharedLock();
	try {
	    SelectionKey key = chan.register(this.selector, SelectionKey.OP_READ);
	    key.attach(adapter);
	    adapter.setKey(key); //set the other side of the association
	    adapter.registered(); //notify the adapter
	} finally {
	    releaseSharedLock();
	}
	log.debug(adapter + " was registered with interest ops " + adapter.getInterestOps() + 
		" " + SelectionKey.OP_READ);
	return adapter;

    }


    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Dispatcher#unregister(java.nio.channels.SelectableChannel)
     */
    //CONCURRENCY: This method is called by any thread who wants to register a channel, but does not need 
    //             special synchronisation since the selector is internally thread-safe. The sharedLock is
    //             taken to make the dispatcher wait for registration completion, before resuming the 
    //             selection
    @Override
    public void unregister(ChannelFacade chan) {
	HandlerAdapter adapter;
	SelectionKey registrationKey;

	if (!(chan instanceof HandlerAdapter)) {
	    throw new IllegalArgumentException ("Not a valid registration token");
	}

	adapter = (HandlerAdapter) chan;
	registrationKey = adapter.key();

	if(registrationKey == null){
	    log.warn("Selection key of " + adapter + " to unregister is null");
	    throw new IllegalArgumentException ("Not a valid registration token, is channel registered?");
	}

	acquireSharedLock();
	try{
	    adapter.unregistering();
	    registrationKey.cancel();	
	} finally {
	    releaseSharedLock();
	}

	adapter.unregistered();
    }



    /**
     * Grab a read lock on the guard object.  A handler thread
     * calls this method when it wants to mutate the state of the
     * Selector.  It must call releaserSharedLock when it is finished,
     * because selection will not resume until all read locks have been
     * released.
     */
    //CONCURRENCY: called by any thread who wishes to register or unregister a channel
    private void acquireSharedLock() {
	guard.readLock().lock();
	log.debug("Shared lock acquired. Now waking up the selector.");
	selector.wakeup(); // if the dispatcher is blocked in a select(), wake him up!
	// (after that it will probably wait on its barrier)
    }

    /**
     * Undo a previous call to acquireSharedLock to indicate that
     * the calling thread no longer needs access to the Selector object.
     */
    //CONCURRENCY: called by any thread who wishes to register or unregister a channel
    private void releaseSharedLock() {
	guard.readLock().unlock();
	log.debug("Shared lock released");
    }


    /**
     * Ask the dispatcher to set the modifications to the status of <code>adapter</code>
     * The <code>adapter</code> must be registered on this dispatcher.
     * <p>
     * Note that changes are not applied immediately but may be deferred after the next selection
     * operation.
     * </p>
     * @param adapter the registered adapter, reflecting the status changes
     */
    //CONCURRENCY: This method is called by any thread who manages an adapter, or by the adapter itself.
    //             Nevertheless the waitingQueue is thread-safe, so there's no need for explicit synchronisation
    public void enqueueStatusChange(HandlerAdapter adapter) {
	boolean interrupted = false;

	Set<SelectionKey> registeredKeys = selector.keys();


	//checking whether the adapter is registered or not
	if(!registeredKeys.contains(adapter.key())) {
	    String msg = "Tried to act on an adaptor that is not registered on the dispatcher";
	    log.error(msg);
	    throw new LowLevelDispatchingException(msg);
	}


	try {
	    while (true) {
		try {
		    waitingForStatusChange.put (adapter);
		    log.debug("Waking up the selector");
		    selector.wakeup(); //wake up the dispatcher if blocked in selection
		    return;
		} catch (InterruptedException e) {
		    interrupted = true;
		}
	    }
	} finally {
	    if (interrupted) Thread.currentThread().interrupt();
	}
    }


    /**
     * Called after a selection operation. Checks whether there are adaptor waiting for a
     * status change. This status change may be a notification of death, or a modification
     * in interestOps.
     */
    private void checkStatusChangeQueue() {
	HandlerAdapter adapter;

	while ((adapter = waitingForStatusChange.poll()) != null) {
	    if (adapter.isDead()) {
		unregister(adapter);
	    } else {
		SelectionKey key = adapter.key();

		if (key.isValid()) {
		    log.debug("interest ops for " + adapter + " are " + adapter.getInterestOps());
		    key.interestOps (adapter.getInterestOps());
		}
	    }
	}

    }

    /**
     * Used to test whether the dispather is dispatching
     * @return true if the dispatcher is in the dispatching cycle
     */
    public boolean isDispatching(){
	return dispatching;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getNumSources() {
	int numSources = 0;
	acquireSharedLock();
	//The selector's keyset is not thread-safe, so I need to synchronize
	try {
	    numSources = selector.keys().size();
	} finally {
	    releaseSharedLock();
	}

	return numSources;
    }


    /**
     * The number of requests currently handled by working threads
     * that were launched by this dispatcher
     * @return the number of requests currently handled by working threads
     *         that were launched by this dispatcher
     */
    public int getNumRequestsServed() {
	return numRequestsServed.get();
    }

    /**
     * Called by the handler adapter only. Used for mngmnt purposes
     */
    protected void requesteServed() {
	numRequestsServed.incrementAndGet(); //increments the number of served requests
    }



    /**
     * Specialised FutureTask class that wraps a HandlerAdapter instance
     * for execution by the thread pool Executor.  This class overrides
     * the done() method to place the contained HandlerAdapter object
     * onto a BlockingQueue for the selection thread to see.
     * 
     * @author Ron Hitchens - original work
     * @author Andrea Reale - adaptation
     * @version 0.0.1    19/set/2009
     */
    private class HandlerFutureTask extends FutureTask<HandlerAdapter>
    {
	private final HandlerAdapter adapter;

	public HandlerFutureTask (HandlerAdapter adapter)
	{
	    super (adapter);
	    this.adapter = adapter;
	}

	protected void done()
	{
	    log.debug("Exectuion of " + adapter + " done. Enqueueing for status change.");
	    enqueueStatusChange (adapter);




	    try {
		// Get result returned by call(), or cause
		// deferred exception to be thrown.  We know
		// the result will be the adapter instance
		// stored above, so we ignore it.
		get();


	    } catch (ExecutionException e) {
		adapter.die();
		log.warn("Handler died", e);
	    } catch (InterruptedException e) {
		Thread.interrupted();
		log.warn("Handler interrupted", e);
	    }
	}
    }


    /* ************************************* */
    /* Listener Registration related methods */
    /* ************************************* */

    @Override
    /**
     * {@inheritDoc}
     */
    //CONCURRENCY: synchronised on the set itself
    public boolean registerListener(DispatcherListener listener) {
	synchronized (registeredListeners) {
	    return registeredListeners.add(listener);
	}
    }

    @Override
    /**
     * {@inheritDoc}
     */
    //CONCURRENCY: synchronised on the set itself
    public boolean removeListener(DispatcherListener listener) {
	synchronized (registeredListeners) {
	    return registeredListeners.remove(listener);
	}
    }




}


